/**
 * LY.com Inc.
 * Copyright (c) 2004-2020 All Rights Reserved.
 */
package com.gitee.kamismile.stone.rpc.common.provider.consumer;

import com.gitee.kamismile.stone.commmon.errorEnum.ErrorEnum;
import com.gitee.kamismile.stone.rpc.common.errorenum.RpcErrorEnum;
import com.gitee.kamismile.stone.rpc.common.exception.RPCException;
import com.gitee.kamismile.stone.rpc.common.provider.producer.StoneProviderMeta;
import com.gitee.kamismile.stone.rpc.common.support.StoneURL;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author dong.li
 * @version $Id: ConsumerMeta, v 0.1 2020/4/2 15:51 dong.li Exp $
 */
public class StoneConsumerMeta {

    //服务    版本,url
    private static ConcurrentHashMap<String, List<StoneURL>> consumerMap = new ConcurrentHashMap<String, List<StoneURL>>();


    private static String appsHashCode ="";

    /**
     * @param serverName
     * @param stoneURL
     */
    public static void addStoneUrl(String serverName, StoneURL stoneURL) {
        List<StoneURL> urlList = consumerMap.get(serverName);

        if (CollectionUtils.isEmpty(urlList)) {
            urlList = new ArrayList<>();
        }

        Optional<StoneURL> optionalStoneURL = urlList.stream()
                .filter(v -> (v.getHost().equals(stoneURL.getHost()) && (v.getPort() == stoneURL.getPort())))
                .findFirst();

        if (optionalStoneURL.isPresent()) {
            StoneURL stoneURLTemp = optionalStoneURL.get();
            urlList.remove(stoneURLTemp);
            if (stoneURLTemp.getWeight() < 50) {
                stoneURL.setWeightAtomic(new AtomicInteger(51));
            } else {
                stoneURL.setWeightAtomic(stoneURLTemp.getWeightAtomic());
            }

            stoneURL.setCallsAtomic(new AtomicInteger(0));
        }

        urlList.add(stoneURL);

        consumerMap.put(serverName, urlList);
    }


    public static synchronized void resetStoneUrls(List<StoneURL> urlList) {
        if (CollectionUtils.isEmpty(urlList)) {
            return;
        }

        Map<String, StoneURL> map = new HashMap<>();
        //有新加入的 老的调用次数重置 为0？ 防止只调用老的

        //老数据删除
        // 低于50得权重 如果从注册中心获取到了 调整到51 再次尝试
        urlList.stream().forEach(v -> {
            addStoneUrl(v.getServerName(), v);
            map.put(v.getServerName() + "`" + v.getHost() + "`" + v.getPort(), v);
        });


        //没有的部分 不是100的 情况权重降到30  防止调用正常但注册中心没有的情况
        //已经低于50的 删除
        Enumeration<String> keys = consumerMap.keys();
        while (keys.hasMoreElements()) {
            String key = keys.nextElement();
            List<StoneURL> list = consumerMap.get(key);

            Iterator<StoneURL> it = list.iterator();
            while (it.hasNext()) {
                StoneURL stoneURL = it.next();
                StoneURL url = map.get(stoneURL.getServerName() + "`" + stoneURL.getHost() + "`" + stoneURL.getPort());
                if (Objects.isNull(url)) {
                    if (stoneURL.getWeight() <= 50) {
                        it.remove();
                    } else {
                        stoneURL.setCallsAtomic(new AtomicInteger(0));
                        if (stoneURL.getWeight() != 100) {
                            stoneURL.setWeightAtomic(new AtomicInteger(35));
                        }else {
                            stoneURL.setWeightAtomic(new AtomicInteger(35));
                        }
                    }
                }
            }
        }

    }

    public static StoneURL findVersion(StoneProviderMeta stoneProviderMeta) {
        String serverName = stoneProviderMeta.getServerName();
        String serverVersion = stoneProviderMeta.getServerVersion();

        List<StoneURL> urls = consumerMap.get(serverName);

        if (CollectionUtils.isEmpty(urls)) {
            appsHashCode =UUID.randomUUID().toString();
            throw new RPCException(RpcErrorEnum.NOTFOUNDSERVER.getCode(), RpcErrorEnum.NOTFOUNDSERVER.getMessage());
        }

        StoneConsumerVersion consumerVersion = new StoneConsumerVersion(serverVersion);

        long numberVersion = consumerVersion.getNumberVersion();
        return getFirstStoneUrl(urls, numberVersion);
    }


    private static StoneURL getFirstStoneUrl(List<StoneURL> urls, long version) {

        Map<Long, List<StoneURL>> urlInfos = urls.stream().filter(v -> v.getWeightAtomic().get() > 50).collect(Collectors.groupingBy(StoneURL::getNumberVersion));
//        Supplier<Stream<StoneURL>> urlStream = () -> urls.stream().sorted(Comparator.comparing(StoneURL::getNumberVersion, Comparator.reverseOrder()).thenComparing(StoneURL::getWeight, Comparator.reverseOrder()).thenComparingLong(StoneURL::getCalls));

        Stream<Long> stoneURLVersionStream = urlInfos.keySet().stream().sorted(Comparator.naturalOrder());

        //永远最新 选择
        if (version == 0) {
            Optional<StoneURL> urlOptional =urlInfos.get(stoneURLVersionStream.findFirst().get()).stream().sorted(Comparator.comparing(StoneURL::getWeight, Comparator.reverseOrder()).thenComparingLong(StoneURL::getCalls)).findFirst();
            if (urlOptional.isPresent()) {
                return urlOptional.get();
            }
        }


        //选择 权重50以上的 最合适的版本
        Optional<Long> aroundNumberVersionOption = stoneURLVersionStream.map(v -> (Math.abs(v - version))).sorted().findFirst();
        if (aroundNumberVersionOption.isPresent()) {
            //根据版本最合适的选择可用率在50以上的
            Optional<StoneURL> stoneURLOptional = chooseStoneURL(version, urlInfos, aroundNumberVersionOption);
            if (stoneURLOptional.isPresent()) {
                return stoneURLOptional.get();
            }

        }
        appsHashCode =UUID.randomUUID().toString();
        throw new RPCException(ErrorEnum.NOSERVICE.getCode(), ErrorEnum.NOSERVICE.getMessage());

    }

    private static Optional<StoneURL> chooseStoneURL(long version, Map<Long, List<StoneURL>> urlInfos, Optional<Long> aroundNumberVersionOption) {
        List<StoneURL> stoneURLS = urlInfos.get(version + aroundNumberVersionOption.get());
        if(CollectionUtils.isEmpty(stoneURLS)){
            stoneURLS = urlInfos.get(version - aroundNumberVersionOption.get());
        }

        Optional<StoneURL> stoneURLAddOptional = stoneURLS.stream().sorted(Comparator.comparing(StoneURL::getWeight, Comparator.reverseOrder()).thenComparingLong(StoneURL::getCalls)).findFirst();
        return stoneURLAddOptional;
    }


    public static String getAppsHashCode() {
        return appsHashCode;
    }
}
